#include "EventPool.h"

void EventPool::init()
{
    epoll_event evt{};
    evt.events = EPOLLIN;
    evt.data.fd = _fd;
    int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, _fd, &evt);
    if (-1 == ret)
    {
        perror("epoll_ctl");
        exit(-1);
    }
}

bool EventPool::isClosed(int cltFd)
{
    char buf[10]{};
    int ret = recv(cltFd, buf, sizeof(buf), MSG_PEEK);
    return (ret == 0);
}

EventPool::EventPool(Acceptor &acceptor)
    : _acceptor(acceptor), _evts(1024), _isLooping{false}
{
    _fd = _acceptor.fd();
    _epfd = epoll_create(1);
    init();
}

EventPool::~EventPool()
{
}

void EventPool::loop()
{
    _isLooping = true;
    while (_isLooping)
    {
        epollWait();
    }
}

void EventPool::unloop()
{
    _isLooping = false;
}

void EventPool::epollWait()
{
    int nready{};
    do
    {
        nready = epoll_wait(_epfd, &*_evts.begin(), _evts.size(), 5000);
        // 如果nready == -1 并且 错误原因是中断，那么就继续执行
    } while (-1 == nready && errno == EINTR);
    if (-1 == nready)
    {
        perror("epoll_wait");
        exit(-1);
    }
    else if (0 == nready)
    {
        cout << "epoll_wait timeout..." << endl;
    }
    // 如果当前ready数量和event_poll大小相等，那么为了处理将来新的链接，扩容event_poll的大小。
    if (nready == _evts.size())
    {
        _evts.resize(_evts.size() * 2);
    }

    for (int idx{}; idx < nready; ++idx)
    {
        // 有新的连接请求
        if (_evts[idx].data.fd == _fd && _evts[idx].events & EPOLLIN)
        {
            int cltFd = _acceptor.accept();
            addNewEpollCtl(cltFd);

            // cout << "cltFd = " << cltFd << endl;
            _tcps.find(cltFd)->second->handleConnect();
        }
        else
        {
            handleMessage(_evts[idx].data.fd);
        }
    }
}

void EventPool::addNewEpollCtl(int cltFd)
{
    epoll_event evt{};
    evt.data.fd = cltFd;
    evt.events = EPOLLIN;
    epoll_ctl(_epfd, EPOLL_CTL_ADD, cltFd, &evt);

    addNewTcpConnect(cltFd);
}

void EventPool::addNewTcpConnect(int cltFd)
{
    auto iter = _tcps.find(cltFd);
    if (iter == _tcps.end())
    {
        TcpConnect *p = new TcpConnect(cltFd);

        // 注册回调函数
        p->setConnectCallback(_handleConnectCallback);
        p->setMessageCallback(_handleMessageCallback);
        p->setClosedCallback(_handleClosedCallback);
        // 注册线程池
        p->setThreadPool(&_pool);

        _tcps.insert(std::make_pair(cltFd, TcpConnectPtr(p)));
    }
}

void EventPool::handleMessage(int cltFd)
{
    // 如果当前客户端断开连接
    auto iter = _tcps.find(cltFd);
    if (isClosed(cltFd))
    {
        delEpollCtl(iter->first);
        delTcpConnect(iter->second);

        _tcps.erase(cltFd);
    }
    else
    {
        iter->second->handleMessage();
    }
}

void EventPool::delEpollCtl(int cltFd)
{
    epoll_ctl(_epfd, EPOLL_CTL_DEL, cltFd, nullptr);
}

void EventPool::delTcpConnect(TcpConnectPtr conn)
{
    conn->handleClosed();
}

void EventPool::setMessageCallback(callback &&cb)
{
    _handleMessageCallback = std::move(cb);
}

void EventPool::setClosedCallback(callback &&cb)
{
    _handleClosedCallback = std::move(cb);
}

void EventPool::setConnectCallback(callback &&cb)
{
    _handleConnectCallback = std::move(cb);
}